package com.registration;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;

public class Register {
    public CuratorFramework curatorFramework;

    public int registerPort;

    private CuratorFramework createWithOptions(String connectionString, RetryPolicy retryPolicy,int connectionTimoutMs,int sessionTimeoutMs){
        return CuratorFrameworkFactory
                .builder()
                .connectString(connectionString)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .build();
    }

    public boolean existNode(String zkPath) throws Exception {
        Stat stat = curatorFramework.checkExists().forPath(zkPath);
        return stat!=null;
    }

    public void createNode(String zkPath,String nodeVal) throws Exception {
        byte[] payload = null;
        payload = nodeVal.getBytes("UTF-8");
        curatorFramework.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath(zkPath,payload);
    }

    public void createSequentialNode(String zkPath,String nodeVal) throws Exception {
        byte[] payload = null;
        payload = nodeVal.getBytes("UTF-8");
        curatorFramework.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                .forPath(zkPath,payload);
    }

    public void createTmpNode(String zkPath,String nodeVal) throws Exception {
        byte[] payload = null;
        payload = nodeVal.getBytes("UTF-8");
        curatorFramework.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL)
                .forPath(zkPath,payload);
    }

    public List<String> retrieveCatalog(String path) throws Exception {
        List<String> children = curatorFramework.getChildren().forPath(path);
        return children;
    }

    public String retrieveVal(String path) throws Exception {
        byte[] arr = curatorFramework.getData().forPath(path);
        return new String(arr,"UTF-8");
    }


    /**
     * 更新节点
     */
    public void updateNode(String zkPath,String data) throws Exception {
        //启动客户端实例,连接服务器
        byte[] payload = data.getBytes("UTF-8");
        curatorFramework.setData()
                .forPath(zkPath, payload);
    }

    public void deleteNode(String zkPath) throws Exception {
            //启动客户端实例,连接服务器
        curatorFramework.delete().forPath(zkPath);
    }

    public void deleteAllNode(String zkPath) throws Exception{
        curatorFramework.delete().deletingChildrenIfNeeded().forPath(zkPath);
    }


    public CuratorCache addWatcher(String workerPath){
        //检查节点是否存在，没有则创建
        CuratorCache cache = null;
        try {
            cache = CuratorCache.build(curatorFramework, workerPath);
            CuratorCache finalCache = cache;
            CuratorCacheListener listener = CuratorCacheListener.builder()
                    .forChanges((oldNode, node) -> {
                        if(node.getData()!=null){
                            System.out.println("old data: " + new String(oldNode.getData()) + " new data:" + new String(node.getData()));
                        }else{
                            System.out.println("节点删除");
                        }
                    })
                    .forCreates((node) -> {
                        System.out.println("节点被创建：" + node.getPath());
                    })
                    .forDeletes((node) -> {
                        System.out.println("节点被删除：" + node.getPath());
                        if(node.getPath().equals(workerPath)){
                            System.out.println("根节点删除，退出监控");
                            finalCache.close();
                        }
                    })
                    .forInitialized(() -> System.out.println("Cache initialized"))
                    .build();
            cache.listenable().addListener(listener);
            cache.start();
            Thread.sleep(1000);
            if(!existNode(workerPath+"/b")) createNode(workerPath+"/b","12");
            Thread.sleep(1000);
            System.out.println(cache.size());
//            deleteNode(workerPath+"/b");
            updateNode(workerPath,"123");
            Thread.sleep(1000);
            updateNode(workerPath,"456");

        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("PathCache监听失败, path="+ workerPath);;
        }
        return cache;
    }

    public void start(int registerPort) throws UnknownHostException {
        this.registerPort = registerPort;
        start(1000,3, InetAddress.getLocalHost().getHostAddress()+":"+registerPort,2000,2000);
    }

    public void start(int retrySleepTimeMs,int maxRetries,String connction,int connectionTimeoutMs,int sessionTimeoutMs){
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(retrySleepTimeMs,maxRetries);
        curatorFramework = createWithOptions(connction,retryPolicy,connectionTimeoutMs,sessionTimeoutMs);
        curatorFramework.start();
    }

    public void close(){
        CloseableUtils.closeQuietly(curatorFramework);
        curatorFramework = null;
    }
}
